package com.citusdata.elven.feeder;

import com.citusdata.elven.datastore.BdbDatastore;
import com.citusdata.elven.datastore.DatastoreEntry;
import com.citusdata.elven.datastore.DatastoreKey;
import com.citusdata.elven.datastore.DatastoreValue;
import com.citusdata.elven.feeder.Log4jHelper;
import com.citusdata.elven.linearRoad.protobuf.LinearRoad.*;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.PrintWriter;
import java.lang.IllegalArgumentException;
import java.util.List;
import org.apache.log4j.Logger;


public class HistoricalDataFeeder {
  static { Log4jHelper.setApplicationName(HistoricalDataFeeder.class); } 
  private static final Logger logger = Logger.getLogger(HistoricalDataFeeder.class);

  private class Fields {
    private Fields() { }

    // Used as array indices, these do not fit into Java's Enum construct.
    public static final int VID   = 0;
    public static final int DAY   = 1;
    public static final int XWAY  = 2;
    public static final int TOLLS = 3;
  }

  private static final String SEPARATOR = ",";
  private static final boolean NEGATIVE_DAYS = true;

  public static TollAssessment parseRequest(String line, boolean negativeDays) 
    throws IOException {
    String[] elements = line.split(SEPARATOR);
    
    TollAssessment.Builder historical = TollAssessment.newBuilder();
    historical.setVid(Integer.valueOf(elements[Fields.VID]));
    historical.setXway(Integer.valueOf(elements[Fields.XWAY]));
    historical.setToll(Integer.valueOf(elements[Fields.TOLLS]));

    if (negativeDays) {
      historical.setTime(Integer.valueOf(elements[Fields.DAY]) * -1);
    } else {
      historical.setTime(Integer.valueOf(elements[Fields.DAY]));
    }

    return (historical.build());
  }


  public static DatastoreEntry parseRequest(String line) throws IOException {
    // Time for historical queries are represented as "negative days", so we
    // always adjust parsed data for historical queries. For other data sets
    // however, such as ones used by unit tests, this adjustment is not needed.  
    TollAssessment histToll = parseRequest(line, NEGATIVE_DAYS);
    String partitionKey = Integer.toString(histToll.getVid());
    long   timestamp    = (long) histToll.getTime();

    DatastoreKey key = new DatastoreKey(partitionKey, timestamp);
    DatastoreValue value = new DatastoreValue(histToll.toByteArray());
    DatastoreEntry entry = new DatastoreEntry(key, value);

    return entry;
  }


  public static boolean readAndValidateEntry(BdbDatastore datastore,
                                             DatastoreKey key, DatastoreValue value) {
    boolean validated = false;

    List testEntries = datastore.getRange(key, key);
    if (!testEntries.isEmpty()) {

      DatastoreEntry testEntry = (DatastoreEntry) testEntries.get(0);
      DatastoreValue testValue = testEntry.getDatastoreValue();

      if (value.equals(testValue)) {
        validated = true;
      }
    }

    return validated;
  }


  public static void main(String[] args) throws IOException {
    if (args.length < 1) {
      logger.fatal("Usage: HistoricalDataFeeder hist_file [database_dir]");
      System.exit(-1);
    }

    BufferedReader historyStream = null;
    BdbDatastore   datastore = null;
    try {
      String historyFileName = args[0];
      historyStream = new BufferedReader(new FileReader(historyFileName));

      // Set default Bdb directory path
      String bdbDirectory = "/var/tmp/bdb";
      if (args.length == 2) {
        bdbDirectory = args[1];
      }

      File directory = new File(bdbDirectory);
      boolean exists  = directory.isDirectory();
      boolean putData = !exists;  // Put data only if the directory does not exist
      if (putData) {
        directory.mkdir();
      } else {
        logger.info("BDB directory already exists, checking for historical values");
      }
      datastore = new BdbDatastore(bdbDirectory);

      int putCount = 0;
      int getCount = 0;
      int validatedCount = 0;
      String inputLine;

      while ((inputLine = historyStream.readLine()) != null) {
        DatastoreEntry request = parseRequest(inputLine);

        DatastoreKey requestKey = request.getDatastoreKey();
        DatastoreValue requestValue = request.getDatastoreValue();

        if (putData) {
          putCount++;
          datastore.put(requestKey, requestValue);
        } else {
          // Check if this entry is extracted properly from the data store
          getCount++;
          boolean validated = readAndValidateEntry(datastore, requestKey, requestValue); 
          if (validated) {
            validatedCount++;
          }
        }
      }

      if (putData) {
        logger.info("Values put into the data store: " + putCount);
      } else {
        logger.info("Values read from the data store: " + getCount +
                    "; values that validated from file: " + validatedCount);
      }

    } finally {
      if (historyStream != null) {
        historyStream.close();
      }

      if (datastore != null) {
        datastore.close();
      }
    }
  }
}